use futures_util::{SinkExt, StreamExt};
use http_body_util::Full;
use hyper::body::Body;
use hyper::body::Bytes;
use hyper::body::Incoming;
use hyper::header::{self, CONNECTION, SEC_WEBSOCKET_ACCEPT, SEC_WEBSOCKET_KEY, UPGRADE};
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper::{body, Request, Response};
use std::convert::Infallible;
use std::net::SocketAddr;
use tokio::net::{TcpListener, TcpStream};
use tokio_tungstenite::{accept_async, accept_hdr_async, tungstenite::Error};
use tracing::Level;
//use tracing::{debug, info, Level};
use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};
use tungstenite::handshake::derive_accept_key;
use tungstenite::handshake::server::Callback;
use tungstenite::handshake::server::ErrorResponse;
use tungstenite::Result;

async fn hello(_: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
    Ok(Response::new(Full::new(Bytes::from("Hello, World!"))))
}

async fn accept_connection(peer: SocketAddr, stream: TcpStream) {
    if let Err(e) = handle_connection(peer, stream).await {
        match e {
            Error::ConnectionClosed | Error::Protocol(_) | Error::Utf8 => (),
            err => eprintln!("Error processing connection: {}", err),
        }
    }
}

async fn handle_connection(peer: SocketAddr, stream: TcpStream) -> Result<()> {
    struct Callb();

    impl Callback for Callb {
        fn on_request(
            self,
            request: &Request<()>,
            response: Response<()>,
        ) -> Result<Response<()>, ErrorResponse> {
            let upgrade = header::HeaderValue::from_static("Upgrade");
            let websocket = header::HeaderValue::from_static("websocket");
            let headers = request.headers();
            let key = headers.get(SEC_WEBSOCKET_KEY);
            let derived = key.map(|k| derive_accept_key(k.as_bytes()));
            for (ref header, _value) in request.headers() {
                println!("* {}", header);
            }
            let mut res = Response::new(());
            res.headers_mut().append(CONNECTION, upgrade);
            res.headers_mut().append(UPGRADE, websocket);
            //res.headers_mut()
            //    .append("Sec-WebSocket-Accept", derived.unwrap().parse().unwrap());
            res.headers_mut()
                .append(SEC_WEBSOCKET_ACCEPT, derived.unwrap().parse().unwrap());
            res.headers_mut().append("hello", "world".parse().unwrap());

            Ok(res)
        }
    }

    let callb_val = Callb {};

    let mut ws_stream = accept_hdr_async(stream, callb_val)
        .await
        .expect("failed to accept");

    //let mut ws_stream = accept_async(stream).await.expect("Failed to accept");

    println!("New WebSocket connection: {}", peer);

    println!("111111111111");
    while let Some(msg) = ws_stream.next().await {
        println!("aaaaaaaaaaaaaaaaa");
        match msg {
            Ok(msg) => {
                println!("Got message: {}", msg);
            }
            Err(e) => {
                println!("Error processing message: {}", e);
            }
        }
    }

    println!("222222222222");

    Ok(())
}

#[tokio::main]
async fn main() {
    //tracing_subscriber::registry().with(fmt::layer()).init();
    tracing_subscriber::fmt().with_max_level(Level::INFO).init();

    let hh = header::SEC_WEBSOCKET_ACCEPT;

    println!("hhh {:?}", hh.as_str());

    //console_subscriber::init();

    let listener2 = TcpListener::bind("localhost:3465").await.unwrap();

    let listener = TcpListener::bind("0.0.0.0:9002").await.unwrap();

    let listener3 = TcpListener::bind("127.0.0.1:9003").await.unwrap();

    loop {
        tokio::select! {
            _ = async {
                while let Ok((_stream, _)) = listener2.accept().await {
                    tokio::spawn(async move { println!("1111 listener") });
                }
            } => {}
            _ = async {
                while let Ok((stream, _)) = listener.accept().await {
                    let peer = stream.peer_addr().expect("connected streams should have a peer address");
                    println!("Peer address: {}", peer);
                    tokio::spawn(accept_connection(peer, stream));
                }
            } => {}
            _ = async {
                while let Ok((stream, _)) = listener3.accept().await {
                    tokio::spawn(async move {
                        if let Err(err) = http1::Builder::new()
                        .serve_connection(stream, service_fn(hello))
                        .await
                        {
                            println!("Error serving connection: {:?}", err);
                        }
                    });
                }
            } => {}
        }
    }
}
